package ods

import java.nio.charset.Charset
import java.util

import com.alibaba.fastjson.JSON
import modes.{AreaInfo, OrderInfo, TableObject}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.conf
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Table}
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}

class HBaseSink extends RichSinkFunction[util.ArrayList[TableObject]]{
  var connection: Connection = _
  var hTable: Table = _
  val tName = "lagou_trade_orders"

  override def open(parameters: Configuration): Unit = {
    val configuration: conf.Configuration = HBaseConfiguration.create()
    configuration.set("hbase.zookeeper.quorum","linux100")
    configuration.set("hbase.zookeeper.property.clientPort", "2181")
    connection = ConnectionFactory.createConnection(configuration)
    hTable = connection.getTable(TableName.valueOf(tName))
  }

  override def close(): Unit = {
    if (hTable != null) {
      hTable.close()
    }

    if (connection != null) {
      connection.close()
    }
  }

  override def invoke(value: util.ArrayList[TableObject], context: SinkFunction.Context[_]): Unit = {
    value.forEach(x => {
      val database: String = x.database
      val tableName: String = x.tableName
      val typeInfo: String = x.typeInfo
      if ("stage9".equalsIgnoreCase(database)){
        if ("lagou_trade_orders".equalsIgnoreCase(tableName)) {
          if ("insert".equalsIgnoreCase(typeInfo)) {
            val dataInfo: OrderInfo = JSON.parseObject(x.dataInfo, classOf[OrderInfo])
            insertTradeOrders(hTable, dataInfo)
          }
        }

        if ("lagou_area".equalsIgnoreCase(tableName)) {
          if ("insert".equalsIgnoreCase(typeInfo)) {
            val areaInfo: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
              insertArea(hTable, areaInfo)
          }
        }
      }
    })
  }

  def insertTradeOrders(hbTable: Table, dataInfo: OrderInfo): Unit = {
    val put: Put = new Put(dataInfo.orderId.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"modifiedTime".getBytes,dataInfo.modifiedTime.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"orderNo".getBytes,dataInfo.orderNo.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"isPay".getBytes,dataInfo.isPay.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"orderId".getBytes,dataInfo.orderId.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"tradeSrc".getBytes,dataInfo.tradeSrc.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"payTime".getBytes,dataInfo.payTime.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"productMoney".getBytes,dataInfo.productMoney.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"totalMoney".getBytes,dataInfo.totalMoney.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"dataFlag".getBytes,dataInfo.dataFlag.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"userId".getBytes,dataInfo.userId.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"areaId".getBytes,dataInfo.areaId.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"createTime".getBytes,dataInfo.createTime.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"payMethod".getBytes,dataInfo.payMethod.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"isRefund".getBytes,dataInfo.isRefund.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"tradeType".getBytes,dataInfo.tradeType.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes,"status".getBytes,dataInfo.status.getBytes(Charset.forName("UTF-8")))
    hbTable.put(put)
  }

  def insertArea(hbTable: Table, areaInfo: AreaInfo): Unit = {
    val put = new Put(areaInfo.id.getBytes())
    put.addColumn("f1".getBytes(), "name".getBytes(), areaInfo.name.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "pid".getBytes(), areaInfo.pid.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "sname".getBytes(), areaInfo.sname.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "level".getBytes(), areaInfo.level.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "citycode".getBytes(), areaInfo.citycode.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "yzcode".getBytes(), areaInfo.yzcode.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "mername".getBytes(), areaInfo.mername.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "lng".getBytes(), areaInfo.Lng.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "lat".getBytes(), areaInfo.Lat.getBytes(Charset.forName("UTF-8")))
    put.addColumn("f1".getBytes(), "pinyin".getBytes(), areaInfo.pinyin.getBytes(Charset.forName("UTF-8")))

    hbTable.put(put)
  }
}
